package com.xhwl.centiotdata.suppliersystem.ganwei;

import java.beans.PropertyDescriptor;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
import java.util.stream.Collectors;

import org.apache.commons.beanutils.PropertyUtilsBean;
import org.apache.http.HttpEntity;
import org.apache.http.HttpResponse;
import org.apache.http.util.EntityUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.xhwl.centiotdata.mq.producer.GeneralProducer;
import com.xhwl.centiotdata.pojo.dto.ganwei.GanweiDeviceDTO;
import com.xhwl.centiotdata.pojo.dto.ganwei.GanweiRomaAlarmDTO;
import com.xhwl.centiotdata.pojo.dto.ganwei.GanweiRomaDataChangedDTO;
import com.xhwl.centiotdata.pojo.dto.ganwei.GanweiRomaDeviceDTO;
import com.xhwl.centiotdata.pojo.dto.ganwei.GanweiRomaDeviceDTO.DeviceInfo;
import com.xhwl.centiotdata.pojo.dto.ganwei.GanweiRomaDeviceDeleteDTO;
import com.xhwl.centiotdata.pojo.dto.ganwei.GanweiRomaOnlineStatusDTO;
import com.xhwl.centiotdata.service.cent.IDeviceCentService;
import com.xhwl.centiotdata.utils.HttpUtils;
import com.xhwl.common.constant.cent.alarm.AlarmConstant;
import com.xhwl.common.dto.cent.device.DeviceDTO;
import com.xhwl.common.enums.device.BaAttrsEnum;
import com.xhwl.common.enums.device.DeviceOnlineStatusEnum;
import com.xhwl.common.exception.BusinessException;
import com.xhwl.common.pojo.cent.alarm.AlarmManagement;
import com.xhwl.common.pojo.cent.device.Device;
import com.xhwl.common.pojo.cent.device.DeviceProjSupplierCategory;
import com.xhwl.common.query.cent.device.DeviceQuery;
import com.xhwl.common.utils.CodeGenerator;
import com.xhwl.common.utils.DateTool;
import com.xhwl.common.utils.MapTool;

import cn.hutool.core.codec.Base64;
import lombok.extern.slf4j.Slf4j;

@Slf4j
@Service
public class GanweiIoTHandler {
	// 每页记录条数
	private static final int PAGE_SIZE = 50;
	private static final ThreadPoolExecutor executor = new ThreadPoolExecutor(50, 100, 3000, TimeUnit.SECONDS,
			new ArrayBlockingQueue<Runnable>(5000), Executors.defaultThreadFactory(),
			new ThreadPoolExecutor.CallerRunsPolicy());
	private static final String ATTRIBUTE_URL = "/ganwei-iot-receive/attribute";
	private static final String EVENT_URL = "/ganwei-iot-receive/event";
	private static final String STATUS_URL = "/ganwei-iot-receive/status";
	private static final String RES_URL = "/ganwei-iot-receive/resp";
	@Autowired
    GeneralProducer generalProducer;
	@Autowired
    private IDeviceCentService deviceService;
	
	
//	public static void main(String[] args) throws Exception {
//		DeviceProjSupplierCategory config = new DeviceProjSupplierCategory();
//		JSONObject properties = new JSONObject();
//		properties.put("topicList", Arrays.asList(
//				"T_IOT_DEVICE_ADDED_GW__sctest1001",
//				"T_IOT_DEVICE_DELETED_GW__sctest1001",
//				"T_IOT_DEVICE_DATACHANGED_GW__sctest1001",
//				"T_IOT_DEVICE_ALARM_GW__sctest1001",
//				"T_IOT_DEVICE_ONLINESTATUS_GW__sctest1001"
//				));
//		properties.put("HWAPPID", "sctest1001__io.ganwei.iot");
//		properties.put("HWAPPKEY", "-@9=7iC//f.P0+ybxa/uAbO+1l%+.01ltwM=y6H7=.==97HPN1RGyi1==3.5Gk%8");
//		properties.put("mqsIpAndPort", "https://192.168.150.61:19292");
//		properties.put("mqsGroupName", "ganwei_iot_ba2");
//		properties.put("romaIpAndPort", "https://192.168.150.213");
//		config.setProperties(properties);
//		config.setProjectId(2238);
//		config.setProductId(21);
//		consumerMsg(config);
//		System.out.println(properties.toJSONString());
//		List<GanweiRomaDeviceDTO> romaDevice = getRomaDevice(config);
//		System.out.println(JSONObject.toJSONString(romaDevice));
//	}
	
	
	public void consumerMsg(DeviceProjSupplierCategory config) throws Exception {
		log.info("开始监听敢为topic config:{}", JSONObject.toJSONString(config));
		long start = System.currentTimeMillis();
		JSONArray topicList = config.getProperties().getJSONArray("topicList");
		if(topicList != null && !topicList.isEmpty()) {
			Map<String, String> headers = new HashMap<String, String>();
			headers.put("Content-Type", "application/json;charset=utf-8");
			headers.put("X-HW-ID", config.getProperties().getString("HWAPPID"));
			headers.put("X-HW-APPKEY", config.getProperties().getString("HWAPPKEY"));
			
			Map<String, String> querys = new HashMap<>();
			querys.put("ack_wait", "300");
			querys.put("time_wait", "5");
			querys.put("max_msgs", "10");
			querys.put("max_bytes", "2097152");
			
			for (Object topicObj : topicList) {
				String topic= (String)topicObj;
				Map<String, String> msgMap = null;
				do {
					msgMap = queryMsg(config, topic, headers, querys);
					if(msgMap != null && !msgMap.isEmpty()) {
						Map<String, String> finalmsgMap = msgMap;
						executor.execute(()->handerMsg(config.getProjectId(), config.getProductId(), new ArrayList<String>(finalmsgMap.values())));
						consumerAck(new ArrayList<>(finalmsgMap.keySet()), config, topic, headers);
					}
				} while (msgMap != null && !msgMap.isEmpty());
			}
		}
		log.info("监听敢为topic结束 耗时:{}秒", (System.currentTimeMillis() -start)/1000.0);
	}
	
	private void handerMsg(Integer projectId, Integer productId, List<String> contentList) {
		try {
			if(contentList != null && !contentList.isEmpty()) {
				List<String> msgList = contentList.stream().map(c->Base64.decodeStr(c)).collect(Collectors.toList());
				if(msgList != null && !msgList.isEmpty()) {
					JSONObject msgJson = JSONObject.parseObject(msgList.get(0));
					if(msgJson != null) {
						String notifyType = msgJson.getString("notifyType");
						if(notifyType == null) {
							return;
						}
						switch (notifyType) {
						case "deviceDataChanged":
							handlerDataChanged(projectId, msgList);
							break;
						case "deviceAdded":
							handlerAddDevice(projectId, productId, msgList);
							break;
						case "deviceDeleted":
							handlerDeleteDevice(projectId, msgList);
							break;
						case "deviceAlarm":
							handlerDeviceAlarm(projectId, msgList);
							break;
						case "onlineStatus":
							handlerOnlineStatus(projectId, msgList);
							break;
						}
					}
				}
			}
		} catch (Exception e) {
			log.error("异步处理敢为topic消息系统异常", e);
		}
	}
	
	
	private void handlerAddDevice(Integer projectId, Integer productId, List<String> msgList) {
		List<GanweiRomaDeviceDTO> dtoList = jsonToBean(msgList, GanweiRomaDeviceDTO.class);
		saveAndUpdateDevice(projectId, productId, dtoList);
	}
	
	private void handlerDeleteDevice(Integer projectId, List<String> msgList) {
		List<GanweiRomaDeviceDeleteDTO> dtoList = jsonToBean(msgList, GanweiRomaDeviceDeleteDTO.class);
		if(dtoList != null && !dtoList.isEmpty()) {
			dtoList.forEach(dto->{
				DeviceQuery deviceQuery = new DeviceQuery();
				deviceQuery.setProjectId(projectId);
				deviceQuery.setIotDeviceId(dto.getDeviceId());
				deviceService.deleteByIotIdV2(deviceQuery);
			});
		}
	}
	
	private void handlerDataChanged(Integer projectId, List<String> msgList) {
		List<GanweiRomaDataChangedDTO> dtoList = jsonToBean(msgList, GanweiRomaDataChangedDTO.class);
		if(dtoList != null && !dtoList.isEmpty()) {
			Map<String, List<GanweiRomaDataChangedDTO>> thirdDeviceMap = dtoList.stream().collect(Collectors.groupingBy(GanweiRomaDataChangedDTO::getDeviceId));
			DeviceQuery deviceQuery = new DeviceQuery();
			deviceQuery.setProjectId(projectId);
			deviceQuery.setThirdDeviceIdList(new ArrayList<>(thirdDeviceMap.keySet()));
			List<DeviceDTO> deviceList = deviceService.list(null, deviceQuery);
			if(deviceList != null && !deviceList.isEmpty()) {
				for (DeviceDTO deviceDTO : deviceList) {
					deviceDTO.setUpdatedAt(new Date());
					JSONObject attributeJson = deviceDTO.getAttributeJson();
					if(attributeJson == null){
						attributeJson = new JSONObject();
						deviceDTO.setAttributeJson(attributeJson);
					}
					attributeJson.put("eventTime", DateTool.getFullDateTime().format(new Date()).replace(" ", "T"));
					List<GanweiRomaDataChangedDTO> dataChangedList = thirdDeviceMap.get(deviceDTO.getThirdDeviceId());
					if(dataChangedList != null && !dataChangedList.isEmpty()) {
						for (GanweiRomaDataChangedDTO dataChangedDTO : dataChangedList) {
							Map<String, Object> map = beanToMap(dataChangedDTO.getService().getData());
							for (Entry<String,Object> entry : map.entrySet()) {
								if(entry.getValue() != null) {
									attributeJson.put(entry.getKey(), entry.getValue());
								}
							}
						}
					}
				}
				deviceService.batchUpdate(null, deviceList);
			}
		}
	}
	
	
	
	private void handlerDeviceAlarm(Integer projectId, List<String> msgList) {
		List<GanweiRomaAlarmDTO> dtoList = jsonToBean(msgList, GanweiRomaAlarmDTO.class);
		if(dtoList != null && !dtoList.isEmpty()) {
			pullAlarm(projectId, dtoList.stream().map(GanweiRomaAlarmDTO::getDeviceId)
					.collect(Collectors.toList()), BaAttrsEnum.GZ.getAlarmCode());
		}
	}
	
	private void handlerOnlineStatus(Integer projectId, List<String> msgList) {
		List<GanweiRomaOnlineStatusDTO> dtoList = jsonToBean(msgList, GanweiRomaOnlineStatusDTO.class);
		if(dtoList != null && !dtoList.isEmpty()) {
			List<Device> deviceList = dtoList.stream().map(dto->{
				Device device = new Device();
				device.setProjectId(projectId);
				device.setThirdDeviceId(dto.getDeviceId());
				if("DraughtFan".equals(dto.getModel()) || "Blower".equals(dto.getModel())) {
					device.setDeviceTypeId(26);
				}
				device.setOnlineStatus("0".equals(dto.getOnlineStatus()) ? DeviceOnlineStatusEnum.ONLINE : DeviceOnlineStatusEnum.OFFLINE);
				device.setUpdatedAt(new Date());
				return device;
			}).collect(Collectors.toList());
			deviceService.asyncUpdateOnlineStatus(deviceList);
			pullAlarm(projectId, deviceList.stream()
					.filter(d->DeviceOnlineStatusEnum.OFFLINE.getCode().equals(d.getOnlineStatus().getCode()))
					.map(Device::getThirdDeviceId).collect(Collectors.toList()), BaAttrsEnum.device.getAlarmCode());
		}
	}
	
	
	private static <T> List<T> jsonToBean(List<String> msgList, Class<T> clazz) {
		if(msgList == null || msgList.isEmpty()) {
			return null;
		}
		List<T> beanList = null;
		try {
			beanList = msgList.stream().map(msg->JSONObject.parseObject(msg, clazz)).collect(Collectors.toList());
		} catch (Exception e) {
			log.info("JSON字符串转Bean对象失败, msg="+msgList, e);
		}
		return beanList;
	}
	
	private Map<String, String> queryMsg(DeviceProjSupplierCategory config, String topic, Map<String, String> headers, Map<String, String> querys) throws Exception {
		Map<String, String> dataMap = new HashMap<>();
 		HttpResponse response = HttpUtils.doGet(config.getProperties().getString("mqsIpAndPort"),
				String.format("/v1/topic/%s/group/%s/messages", topic, config.getProperties().getString("mqsGroupName")),
				null, headers, querys);
		if (200 != response.getStatusLine().getStatusCode()) {
			throw new BusinessException("请求敢为IOT-ROMA平台系统异常:code=" + response.getStatusLine().getStatusCode());
		}
		HttpEntity resEntity = response.getEntity();
		String resp = null;
		if (resEntity != null) {
			resp = EntityUtils.toString(resEntity, "UTF-8");
//			log.info("查询敢为IOT-ROMA平台topic消息 topic:{}, resp: {}...", topic, StringUtils.substring(resp, 0, 64));
			JSONArray jsonArray = JSON.parseArray(resp);
			if (jsonArray != null && !jsonArray.isEmpty()) {
				for (Object objJson : jsonArray) {
					JSONObject jsonObj = (JSONObject) objJson;
					if (jsonObj != null) {
						String handler = jsonObj.getString("handler");
						JSONObject messageJson = jsonObj.getJSONObject("message");
						if(messageJson != null) {
							String content = messageJson.getString("content");
							dataMap.put(handler, content);
						}
					}
				}
			}
		}
		return dataMap;
	}
	
	
	private static void consumerAck(List<String> handlerList, DeviceProjSupplierCategory config, String topic, Map<String, String> headers) throws Exception {
		if(handlerList == null || handlerList.isEmpty()) {
			return;
		}
		Map<String, Object> body = new HashMap<>();
		body.put("messages", handlerList.stream().map(h-> MapTool.getSOMap("status", "success", "handler", h)).collect(Collectors.toList()));
		HttpResponse response = HttpUtils.doPost(config.getProperties().getString("mqsIpAndPort"),
				String.format("/v1/topic/%s/group/%s/messages", topic, config.getProperties().getString("mqsGroupName")),
				null, headers, new HashMap<String, String>(),
				JSONObject.toJSONString(body));
		if (200 != response.getStatusLine().getStatusCode()) {
			throw new BusinessException("请求敢为IOT-ROMA平台系统异常:code=" + response.getStatusLine().getStatusCode());
		}
//		HttpEntity resEntity = response.getEntity();
//		if (resEntity != null) {
//			String resp = EntityUtils.toString(resEntity, "UTF-8");
//			log.info("敢为IOT-ROMA平台topic消息结果:resp{}", resp);
//		}
	}

	
	/**
	 * 通过ROMA平台查询敢为IoT平台的设备列表 (项目上线时执行一次即可)
	 * @param config
	 * @return
	 * @throws Exception
	 */
	public List<GanweiRomaDeviceDTO> getRomaDevice(DeviceProjSupplierCategory config) throws Exception {
		List<GanweiRomaDeviceDTO> dataList = new ArrayList<>();
		
		Map<String, String> headers = new HashMap<String, String>();
		headers.put("Content-Type", "application/x-www-form-urlencoded; charset=UTF-8");
		headers.put("X-HW-ID", config.getProperties().getString("HWAPPID"));
		headers.put("X-HW-APPKEY", config.getProperties().getString("HWAPPKEY"));
		
		Map<String, String> querys = new HashMap<>();
		querys.put("pagesize", String.valueOf(PAGE_SIZE));
		Integer totalCount = 0;
		int pageNum = 1;
		do {
			querys.put("pageNo", String.valueOf(pageNum));
			HttpResponse response = HttpUtils.doGet(config.getProperties().getString("romaIpAndPort"),
					"/io/sctest1001__io.ganwei.iot/south/device-management/v1.1/deviceList", null, headers, querys);
			if (200 != response.getStatusLine().getStatusCode()) {
				throw new BusinessException("请求敢为IOT-ROMA平台系统异常:code=" + response.getStatusLine().getStatusCode());
			}
			HttpEntity resEntity = response.getEntity();
			String resp = null;
			if (resEntity != null) {
				resp = EntityUtils.toString(resEntity, "UTF-8");
				log.info("查询敢为IOT-ROMA平台设备列表:resp{}", resp);
				JSONObject jsonObj = JSON.parseObject(resp);
				if (jsonObj != null && "0".equals(jsonObj.getString("resCode"))) {
					JSONObject resultJson = jsonObj.getJSONObject("result");
					if (resultJson != null) {
						totalCount = resultJson.getInteger("totalCount");
						JSONArray dataJsonArray = resultJson.getJSONArray("devices");
						if (dataJsonArray != null && !dataJsonArray.isEmpty()) {
							for (Object objJson : dataJsonArray) {
								JSONObject jsonOb = (JSONObject) objJson;
								if (jsonOb != null) {
									GanweiRomaDeviceDTO dto = jsonOb.toJavaObject(GanweiRomaDeviceDTO.class);
									dataList.add(dto);
								}
							}
						}
					}
				}
			}
		} while (totalCount > pageNum++ * PAGE_SIZE);
		executor.execute(() -> saveAndUpdateDevice(config.getProjectId(), config.getProductId(), dataList));
		return dataList;
	}
	
	
	public void saveAndUpdateDevice(Integer projectId, Integer productId, List<GanweiRomaDeviceDTO> dtoList) {
		if(dtoList == null || dtoList.isEmpty()) {
			return;
		}
		Map<String, GanweiRomaDeviceDTO> thirdDeviceMap = dtoList.stream().collect(
				Collectors.toMap(GanweiRomaDeviceDTO::getDeviceId, Function.identity(), (v1, v2) -> v2));
		List<Device> addList = new ArrayList<>();
		List<DeviceDTO> updateList = new ArrayList<>();
		DeviceQuery deviceQuery = new DeviceQuery();
		deviceQuery.setProjectId(projectId);
		deviceQuery.setThirdDeviceIdList(new ArrayList<>(thirdDeviceMap.keySet()));
		List<DeviceDTO> deviceList = deviceService.list(null, deviceQuery);
		if(deviceList != null && !deviceList.isEmpty()) {
			for (DeviceDTO deviceDTO : deviceList) {
				GanweiRomaDeviceDTO updateDevice = thirdDeviceMap.remove(deviceDTO.getThirdDeviceId());
				if(updateDevice != null) {
					updateDevice(updateDevice, deviceDTO);
					updateList.add(deviceDTO);
				}
			}
		}
		if(!thirdDeviceMap.isEmpty()) {
			for (Entry<String, GanweiRomaDeviceDTO> entry : thirdDeviceMap.entrySet()) {
				Device device = saveDevice(entry.getValue());
	        	if(device != null) {
	        		device.setProjectId(projectId);
	        		device.setProductId(productId);
	        		addList.add(device);
	        	}
			}
		}
		if(!addList.isEmpty()) {
			deviceService.batchSave(null, addList);
		}
		if(!updateList.isEmpty()) {
			deviceService.batchUpdate(null, updateList);
			pullAlarm(projectId, updateList.stream()
					.filter(d->Integer.valueOf(DeviceOnlineStatusEnum.OFFLINE.getId()).equals(d.getOnlineStatus()))
					.map(DeviceDTO::getThirdDeviceId).collect(Collectors.toList()), BaAttrsEnum.device.getAlarmCode());
		}
	}
	
	private Device saveDevice(GanweiRomaDeviceDTO dto) {
		Device device = new Device();
//		device.setProjectId(projectId);
		device.setIotDeviceId(dto.getDeviceId());
		device.setThirdDeviceId(dto.getDeviceId());
//		device.setProductId(productId);
		if(dto.getDeviceInfo() != null) {
			DeviceInfo deviceInfo = dto.getDeviceInfo();
			String deviceType = deviceInfo.getDeviceType();
			if("DraughtFan".equals(deviceType) || "Blower".equals(deviceType)) {
				//风机
				device.setCategoryId(11);
				device.setDeviceTypeId(26);
			}else{
				return null;
			}
			device.setLongitude(deviceInfo.getLongitude());
			device.setLatitude(deviceInfo.getLatitude());
			device.setNodeId(deviceInfo.getNodeId());
			device.setName(deviceInfo.getName());
			device.setDetailPosition(deviceInfo.getName());
			device.setDeviceStatus(1);
			device.setOnlineStatus(DeviceOnlineStatusEnum.ONLINE.getCode().equals(deviceInfo.getStatus())
					?DeviceOnlineStatusEnum.ONLINE:DeviceOnlineStatusEnum.OFFLINE);
		}
		return device;
	}
	
	private void updateDevice( GanweiRomaDeviceDTO dto, DeviceDTO deviceDTO) {
		deviceDTO.setIotDeviceId(dto.getDeviceId());
		deviceDTO.setThirdDeviceId(dto.getDeviceId());
		if(dto.getDeviceInfo() != null) {
			DeviceInfo deviceInfo = dto.getDeviceInfo();
			String deviceType = deviceInfo.getDeviceType();
			if("DraughtFan".equals(deviceType) || "Blower".equals(deviceType)) {
				//风机
				deviceDTO.setCategoryId(11);
				deviceDTO.setDeviceTypeId(26);
			}else{
				return;
			}
			deviceDTO.setLongitude(deviceInfo.getLongitude());
			deviceDTO.setLatitude(deviceInfo.getLatitude());
			deviceDTO.setNodeId(deviceInfo.getNodeId());
			deviceDTO.setName(deviceInfo.getName());
			deviceDTO.setDetailPosition(deviceInfo.getName());
			deviceDTO.setDeviceStatus(1);
			deviceDTO.setOnlineStatus(DeviceOnlineStatusEnum.ONLINE.getCode().equals(deviceInfo.getStatus())
					?DeviceOnlineStatusEnum.ONLINE.getId():DeviceOnlineStatusEnum.OFFLINE.getId());
		}
	}
	/**
	 * 查询敢为IoT平台设备列表
	 * 
	 * @param config
	 * @return
	 * @throws Exception
	 */
	public List<GanweiDeviceDTO> getDevice(DeviceProjSupplierCategory config) throws Exception {
		List<GanweiDeviceDTO> dataList = new ArrayList<>();
		List<String> terminalTypeIdList = new ArrayList<>();
		Map<String, Object> body = new HashMap<>();
		body.put("pagesize", PAGE_SIZE);
		Integer totalPage = 0;
		int pageNum = 1;
		do {
			body.put("pageNo", pageNum);
			HttpResponse response = HttpUtils.doPost(config.getProperties().getString("ipAndPort"),
					"/iot/ganwei/v2/devices", null, getHeaders(config, true), new HashMap<String, String>(),
					JSONObject.toJSONString(body));
			if (200 != response.getStatusLine().getStatusCode()) {
				throw new BusinessException("请求敢为IOT平台系统异常:code=" + response.getStatusLine().getStatusCode());
			}
			HttpEntity resEntity = response.getEntity();
			String res = null;
			if (resEntity != null) {
				res = EntityUtils.toString(resEntity, "UTF-8");
				log.info("查询敢为IOT平台设备列表:res{}", res);
				JSONObject jsonObj = JSON.parseObject(res);
				if (jsonObj != null && "0".equals(jsonObj.getString("code"))) {
					JSONObject dataJson = jsonObj.getJSONObject("data");
					if (dataJson != null) {
						totalPage = dataJson.getInteger("totalPage");
						JSONArray dataJsonArray = dataJson.getJSONArray("list");
						if (dataJsonArray != null && !dataJsonArray.isEmpty()) {
							for (Object objJson : dataJsonArray) {
								JSONObject jsonOb = (JSONObject) objJson;
								if (jsonOb != null) {
									GanweiDeviceDTO dto = jsonOb.toJavaObject(GanweiDeviceDTO.class);
									terminalTypeIdList.add(dto.getTerminalTypeIds());
									dataList.add(dto);
								}
							}
						}
					}
				}
			}
		} while (totalPage > pageNum++);
		executor.execute(() -> executeSubscription(config, terminalTypeIdList));
		return dataList;
	}

	/**
	 * 查询敢为IoT平台单个终端设备信息
	 * 
	 * @param config
	 * @param terminalId
	 * @return
	 * @throws Exception
	 */
	public GanweiDeviceDTO getDeviceInfo(DeviceProjSupplierCategory config, int terminalId) throws Exception {
		Map<String, Object> body = new HashMap<>();
		body.put("terminalId", terminalId);
		HttpResponse response = HttpUtils.doPost(config.getProperties().getString("ipAndPort"), "/iot/ganwei/v2/device",
				null, getHeaders(config, true), new HashMap<String, String>(), JSONObject.toJSONString(body));
		if (200 != response.getStatusLine().getStatusCode()) {
			throw new BusinessException("请求敢为IOT平台系统异常:code=" + response.getStatusLine().getStatusCode());
		}
		HttpEntity resEntity = response.getEntity();
		String res = null;
		if (resEntity != null) {
			res = EntityUtils.toString(resEntity, "UTF-8");
			log.info("查询敢为IOT平台设备详情:res{}", res);
			JSONObject jsonObj = JSON.parseObject(res);
			if (jsonObj != null && "0".equals(jsonObj.getString("code"))) {
				JSONObject dataJson = jsonObj.getJSONObject("data");
				if (dataJson != null) {
					return dataJson.toJavaObject(GanweiDeviceDTO.class);
				}
			}
		}
		return null;
	}

	/**
	 * 查询敢为IoT平台设备类型
	 * 
	 * @param config
	 * @return
	 * @throws Exception
	 */
	public Map<String, String> getDeviceType(DeviceProjSupplierCategory config) throws Exception {
		Map<String, String> dataMap = new HashMap<>();
		HttpResponse response = HttpUtils.doPost(config.getProperties().getString("ipAndPort"),
				"/iot/ganwei/v2/device/types", null, getHeaders(config, true), new HashMap<String, String>(),
				JSONObject.toJSONString(new HashMap<>()));
		if (200 != response.getStatusLine().getStatusCode()) {
			throw new BusinessException("请求敢为IOT平台系统异常:code=" + response.getStatusLine().getStatusCode());
		}
		HttpEntity resEntity = response.getEntity();
		String res = null;
		if (resEntity != null) {
			res = EntityUtils.toString(resEntity, "UTF-8");
			log.info("查询敢为IOT平台设备类型列表:res{}", res);
			JSONObject jsonObj = JSON.parseObject(res);
			if (jsonObj != null && "0".equals(jsonObj.getString("code"))) {
				JSONArray dataJsonArray = jsonObj.getJSONArray("data");
				if (dataJsonArray != null && !dataJsonArray.isEmpty()) {
					for (Object objJson : dataJsonArray) {
						JSONObject jsonOb = (JSONObject) objJson;
						if (jsonOb != null) {
							dataMap.put(jsonOb.getString("terminalTypeId"), jsonOb.getString("terminalTypeName"));
						}
					}
				}
			}
		}
		return dataMap;
	}

	/**
	 * 终端指令下发
	 * 
	 * @param config
	 * @param terminalId
	 * @param propertyCode
	 * @param value
	 * @return
	 * @throws Exception
	 */
	public Boolean command(DeviceProjSupplierCategory config, int terminalId, String propertyCode, String value)
			throws Exception {
		Map<String, Object> body = new HashMap<>();
		body.put("terminalId", terminalId);
		body.put("propertyCode", propertyCode);
		body.put("value", value);
		HttpResponse response = HttpUtils.doPost(config.getProperties().getString("ipAndPort"),
				"/iot/ganwei/v2/device/property/command", null, getHeaders(config, true), new HashMap<String, String>(),
				JSONObject.toJSONString(body));
		if (200 != response.getStatusLine().getStatusCode()) {
			throw new BusinessException("请求敢为IOT平台系统异常:code=" + response.getStatusLine().getStatusCode());
		}
		HttpEntity resEntity = response.getEntity();
		String res = null;
		if (resEntity != null) {
			res = EntityUtils.toString(resEntity, "UTF-8");
			log.info("敢为IOT平台终端指令下发:res{}", res);
			JSONObject jsonObj = JSON.parseObject(res);
			if (jsonObj != null && "200".equals(jsonObj.getString("code"))) {
				return Boolean.TRUE;
			}
		}
		return Boolean.FALSE;
	}

	/**
	 * 执行订阅事件
	 * 
	 * @param config
	 * @param terminalTypeIdList
	 */
	private void executeSubscription(DeviceProjSupplierCategory config, List<String> terminalTypeIdList) {
		try {
			List<Integer> dataTypeList = Arrays.asList(1, 2, 3, 4);
			Map<Integer, Integer> subscriptionMap = getSubscriptionList(config);
			for (Integer dataType : dataTypeList) {
				Integer subscriptionId = subscriptionMap.get(dataType);
				if (subscriptionId != null) {
					editSubscription(config, terminalTypeIdList, dataType, subscriptionId);
				} else {
					addSubscription(config, terminalTypeIdList, dataType);
				}
			}
		} catch (Exception e) {
			log.error("执行敢为IoT平台订阅事件系统异常", e);
		}
	}

	/**
	 * 创建订阅
	 * 
	 * @param config
	 * @param terminalTypeIdList
	 * @param dataType
	 * @return
	 * @throws Exception
	 */
	private Integer addSubscription(DeviceProjSupplierCategory config, List<String> terminalTypeIdList,
			int dataType) throws Exception {
		Map<String, Object> body = new HashMap<>();
		setSubscriptionType(dataType, config.getProperties().getString("callbackUrl"), body);
		body.put("terminalTypeIdList", terminalTypeIdList);
		body.put("callbackappId", config.getProperties().getString("callbackappId"));
		body.put("callbackappkey", config.getProperties().getString("callbackappkey"));
		body.put("protocolType", 2);
		if (body.get("callbackUrl") != null && String.valueOf(body.get("callbackUrl")).startsWith("https")) {
			body.put("protocolType", 1);
		}
		body.put("dataType", dataType);
		body.put("state", 1);
		HttpResponse response = HttpUtils.doPost(config.getProperties().getString("ipAndPort"),
				"/iot/ganwei/v2/subscription/create", null, getHeaders(config, true), new HashMap<String, String>(),
				JSONObject.toJSONString(body));
		if (200 != response.getStatusLine().getStatusCode()) {
			throw new BusinessException("请求敢为IOT平台系统异常:code=" + response.getStatusLine().getStatusCode());
		}
		HttpEntity resEntity = response.getEntity();
		String res = null;
		if (resEntity != null) {
			res = EntityUtils.toString(resEntity, "UTF-8");
			log.info("创建敢为IOT平台订阅事件:res{}", res);
			JSONObject jsonObj = JSON.parseObject(res);
			if (jsonObj != null && "0".equals(jsonObj.getString("code"))) {
				JSONObject dataJson = jsonObj.getJSONObject("data");
				if (dataJson != null) {
					return dataJson.getInteger("Id");
				}
			}
		}
		throw new BusinessException("创建敢为IOT平台订阅事件失败, res:" + res);
	}

	/**
	 * 订阅事件类型
	 * 
	 * @param dataType
	 * @param callbackUrl
	 * @param body
	 */
	private void setSubscriptionType(int dataType, String callbackUrl, Map<String, Object> body) {
		switch (dataType) {
		case 1:
			body.put("subscriptionName", CodeGenerator.getCode("终端属性变化"));
			body.put("callbackUrl", callbackUrl + ATTRIBUTE_URL);
			break;
		case 2:
			body.put("subscriptionName", CodeGenerator.getCode("终端事件"));
			body.put("callbackUrl", callbackUrl + EVENT_URL);
			break;
		case 3:
			body.put("subscriptionName", CodeGenerator.getCode("终端状态变化"));
			body.put("callbackUrl", callbackUrl + STATUS_URL);
			break;
		case 4:
			body.put("subscriptionName", CodeGenerator.getCode("终端命令响应"));
			body.put("callbackUrl", callbackUrl + RES_URL);
			break;
		}
	}

	/**
	 * 查询订阅列表
	 * 
	 * @param config
	 * @return
	 * @throws Exception
	 */
	private Map<Integer, Integer> getSubscriptionList(DeviceProjSupplierCategory config) throws Exception {
		Map<Integer, Integer> dataMap = new HashMap<>();
		Map<String, Object> body = new HashMap<>();
		body.put("pagesize", PAGE_SIZE);
		body.put("state", 1);
		Integer totalPage = 0;
		int pageNum = 1;
		do {
			body.put("pageNo", pageNum);
			HttpResponse response = HttpUtils.doPost(config.getProperties().getString("ipAndPort"),
					"/iot/ganwei/v2/subscription/query", null, getHeaders(config, true), new HashMap<String, String>(),
					JSONObject.toJSONString(body));
			if (200 != response.getStatusLine().getStatusCode()) {
				throw new BusinessException("请求敢为IOT平台系统异常:code=" + response.getStatusLine().getStatusCode());
			}
			HttpEntity resEntity = response.getEntity();
			String res = null;
			if (resEntity != null) {
				res = EntityUtils.toString(resEntity, "UTF-8");
				log.info("查询敢为IOT平台订阅列表:res{}", res);
				JSONObject jsonObj = JSON.parseObject(res);
				if (jsonObj != null && "0".equals(jsonObj.getString("code"))) {
					JSONObject dataJson = jsonObj.getJSONObject("data");
					if (dataJson != null) {
						totalPage = dataJson.getInteger("totalPage");
						JSONArray dataJsonArray = dataJson.getJSONArray("list");
						if (dataJsonArray != null && !dataJsonArray.isEmpty()) {
							for (Object objJson : dataJsonArray) {
								JSONObject jsonOb = (JSONObject) objJson;
								if (jsonOb != null) {
									Integer subscriptionId = dataMap.put(jsonOb.getInteger("dataType"),
											jsonOb.getInteger("subscriptionId"));
									if (subscriptionId != null) {
										deleteSubscription(config, subscriptionId);
									}
								}
							}
						}
					}
				}
			}
		} while (totalPage > pageNum++);
		return dataMap;
	}

	/**
	 * 编辑订阅
	 * 
	 * @param config
	 * @param terminalTypeIdList
	 * @param dataType
	 * @param subscriptionId
	 * @return
	 * @throws Exception
	 */
	private Integer editSubscription(DeviceProjSupplierCategory config, List<String> terminalTypeIdList,
			int dataType, int subscriptionId) throws Exception {
		Map<String, Object> body = new HashMap<>();
		body.put("subscriptionId", subscriptionId);
		setSubscriptionType(dataType, config.getProperties().getString("callbackUrl"), body);
		body.put("terminalTypeIdList", terminalTypeIdList);
		body.put("callbackappId", config.getProperties().getString("callbackappId"));
		body.put("callbackappkey", config.getProperties().getString("callbackappkey"));
		body.put("protocolType", 2);
		if (body.get("callbackUrl") != null && String.valueOf(body.get("callbackUrl")).startsWith("https")) {
			body.put("protocolType", 1);
		}
		body.put("dataType", dataType);
		body.put("state", 1);
		HttpResponse response = HttpUtils.doPost(config.getProperties().getString("ipAndPort"),
				"/iot/ganwei/v2/subscription/edit", null, getHeaders(config, true), new HashMap<String, String>(),
				JSONObject.toJSONString(body));
		if (200 != response.getStatusLine().getStatusCode()) {
			throw new BusinessException("请求敢为IOT平台系统异常:code=" + response.getStatusLine().getStatusCode());
		}
		HttpEntity resEntity = response.getEntity();
		String res = null;
		if (resEntity != null) {
			res = EntityUtils.toString(resEntity, "UTF-8");
			log.info("修改敢为IOT平台订阅事件:res{}", res);
			JSONObject jsonObj = JSON.parseObject(res);
			if (jsonObj != null && "0".equals(jsonObj.getString("code"))) {
				JSONObject dataJson = jsonObj.getJSONObject("data");
				if (dataJson != null) {
					return dataJson.getInteger("Id");
				}
			}
		}
		throw new BusinessException("修改敢为IOT平台订阅事件失败, res:" + res);
	}

	/**
	 * 删除订阅
	 * 
	 * @param config
	 * @param subscriptionId
	 * @return
	 * @throws Exception
	 */
	private Boolean deleteSubscription(DeviceProjSupplierCategory config, int subscriptionId) throws Exception {
		Map<String, Object> body = new HashMap<>();
		body.put("subscriptionId", subscriptionId);
		HttpResponse response = HttpUtils.doPost(config.getProperties().getString("ipAndPort"),
				"/iot/ganwei/v2/subscription/delete", null, getHeaders(config, true), new HashMap<String, String>(),
				JSONObject.toJSONString(body));
		if (200 != response.getStatusLine().getStatusCode()) {
			throw new BusinessException("请求敢为IOT平台系统异常:code=" + response.getStatusLine().getStatusCode());
		}
		HttpEntity resEntity = response.getEntity();
		String res = null;
		if (resEntity != null) {
			res = EntityUtils.toString(resEntity, "UTF-8");
			log.info("删除敢为IOT平台订阅事件:res{}", res);
			JSONObject jsonObj = JSON.parseObject(res);
			if (jsonObj != null && "0".equals(jsonObj.getString("code"))) {
				return Boolean.TRUE;
			}
		}
		throw new BusinessException("删除敢为IOT平台订阅事件失败, res:" + res);
	}


	/**
	 * 鉴权
	 * 
	 * @param config
	 * @return
	 * @throws Exception
	 */
	private String getToken(DeviceProjSupplierCategory config) throws Exception {
		if (config.getProperties() == null) {
			throw new BusinessException("集成配置错误,properties为空, 跳过任务:" + JSONObject.toJSONString(config));
		}
		Map<String, String> bodys = new HashMap<>();
		bodys.put("appId", config.getProperties().getString("appId"));
		bodys.put("secret", config.getProperties().getString("appSecret"));
		HttpResponse response = HttpUtils.doPost(config.getProperties().getString("ipAndPort"),
				"/iot/ganwei/v2/auth/getToken", null, getHeaders(config, false), new HashMap<String, String>(),
				JSONObject.toJSONString(bodys));
		if (200 != response.getStatusLine().getStatusCode()) {
			throw new BusinessException("请求敢为IOT平台系统异常:code=" + response.getStatusLine().getStatusCode());
		}
		HttpEntity resEntity = response.getEntity();
		String res = null;
		if (resEntity != null) {
			res = EntityUtils.toString(resEntity, "UTF-8");
			log.info("敢为IOT平台鉴权:res{}", res);
			JSONObject jsonObject = JSON.parseObject(res);
			if (jsonObject != null && "0".equals(jsonObject.getString("code"))) {
				JSONObject dataJson = jsonObject.getJSONObject("data");
				if (dataJson != null) {
					return dataJson.getString("accessToken");
				}
			}
		}
		throw new BusinessException("请求敢为IOT平台获取access_token失败, res:" + res);
	}

	/**
	 * 敢为iot平台accessToken
	 * 
	 * @param config
	 * @param tokenFlag
	 * @return
	 * @throws Exception
	 */
	private Map<String, String> getHeaders(DeviceProjSupplierCategory config, boolean tokenFlag)
			throws Exception {
		Map<String, String> headers = new HashMap<String, String>();
		headers.put("Content-Type", "application/json; charset=utf-8");
		headers.put("X-ClientId", config.getProperties().getString("appId"));
		if (tokenFlag) {
			headers.put("Authorization", "Bearer " + getToken(config));
		}
		return headers;
	}
	
	
	private static Map<String, Object> beanToMap(Object obj) {
        if(obj == null){
            return null;
        }
        Map<String, Object> params = new HashMap<>();
        try {
            PropertyUtilsBean propertyUtilsBean = new PropertyUtilsBean();
            PropertyDescriptor[] descriptors = propertyUtilsBean.getPropertyDescriptors(obj);
            for (int i = 0; i < descriptors.length; i++) {
                String name = descriptors[i].getName();
                if (!"class".equals(name)) {
                    params.put(name, propertyUtilsBean.getNestedProperty(obj, name));
                }
            }
        } catch (Exception e) {
        }
        return params;
    }
	
	private void pullAlarm(Integer projectId, List<String> thirdDeviceIdList, String alarmCode) {
		if(thirdDeviceIdList == null || thirdDeviceIdList.isEmpty()) {
			return;
		}
		DeviceQuery deviceQuery = new DeviceQuery();
		deviceQuery.setProjectId(projectId);
		deviceQuery.setThirdDeviceIdList(thirdDeviceIdList);
		List<DeviceDTO> deviceList = deviceService.list(null, deviceQuery);
		if(deviceList != null && !deviceList.isEmpty()) {
			deviceList.stream().forEach(a->sendAlarm(projectId, a.getId(), alarmCode));
		}
	}
	
	/**
     * 发送告警
     *
     * @param projectId
     * @param deviceId  中台设备id
     * @param alarmCode 告警code
     */
    private void sendAlarm(Integer projectId, Integer deviceId, String alarmCode) {
        AlarmManagement alarmManagement = new AlarmManagement();
        //组装公共部分数据
        alarmManagement.setProjectId(projectId);
        alarmManagement.setAlarmTime(new Date());
        alarmManagement.setDeviceId(deviceId);
        alarmManagement.setAlarmInventoryCode(alarmCode);
//        log.info("发出告警: {}", JSON.toJSONString(alarmManagement));
        generalProducer.send(AlarmConstant.ROMA_TOPIC, alarmManagement);
    }
    
}
